package avicit.bdp.dds.api.utils.notebook;

import avicit.bdp.common.utils.ConfigUtils;
import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.dds.api.dto.NoteBookDTO;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.thread.ThreadUtils;
import avicit.bdp.dds.common.utils.OSUtils;
import avicit.platform6.core.exception.BusinessException;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.jvnet.winp.WinProcess;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;

/**
 * @金航数码科技有限责任公司
 * @作者：developer
 * @邮箱：developer@avic-digital.com
 * @创建时间： 2022-05-23
 * @类说明：NoteBookExecuteThread
 * @修改记录：
 * @注意事项：
 * @主要功能：NoteBook后台任务
 */
public class NoteBookExecuteThread implements Runnable {
    private static final Logger logger = LoggerFactory.getLogger(NoteBookExecuteThread.class);

    /**
     * For Unix-like, using sh
     */
    public static final String SH = "sh";

    /**
     * For Windows, using cmd.exe
     */
    public static final String CMD = "cmd.exe";

    private NoteBookDTO noteBookDTO;

    /**
     * process
     */
    private Process process;

    /**
     * 任务执行上下文
     */
    private NoteBookTaskContext taskContext;

    /**
     * 任务上下文缓存
     */
    private NoteBookTaskCacheManager noteBookTaskCacheManager;

    /**
     * NoteBook服务默认用户路径
     */
    private String baseDir;

    /**
     * 无参构造函数
     */
    public NoteBookExecuteThread() {
    }

    /**
     * 有参构造函数
     * @param noteBookDTO
     */
    public NoteBookExecuteThread(NoteBookDTO noteBookDTO) {
        this.noteBookDTO = noteBookDTO;
        this.taskContext = new NoteBookTaskContext();
        this.noteBookTaskCacheManager = SpringApplicationContext.getBean(NoteBookTaskCacheManager.class);
        this.baseDir = getNoteBookBaseDir();
    }

    @Override
    public void run() {
        try {
            makeUserDir(this.noteBookDTO.getBaseUrl());
            List<String> commands = buildCommand();
            runCommand(commands);
        } catch (Exception e) {
            throw new BusinessException(e.getMessage());
        }
    }

    /**
     * 构建Jupyter NoteBook执行命令
     * @return
     */
    private List<String> buildCommand() {
        List<String> commands = new ArrayList<>();

        commands.add("jupyter");
        commands.add("lab");
        commands.add("--port");
        commands.add(String.format("%s", this.noteBookDTO.getPort()));
        commands.add("--allow-root");
        commands.add("--notebook-dir=" + this.baseDir + "/" + this.noteBookDTO.getBaseUrl());
        commands.add("--no-browser");
        commands.add("--ServerApp.base_url=" + this.noteBookDTO.getBaseUrl());
        commands.add("--ServerApp.token=" + this.noteBookDTO.getToken());
        commands.add("--ServerApp.allow_origin=\"*\"");
        commands.add("--ServerApp.ip=\"*\"");

        return commands;
    }

    /**
     * 拉起Jupyter NoteBook后台进程
     * @param commands
     */
    private void runCommand(List<String> commands) throws Exception {
        if (CollectionUtils.isEmpty(commands)) {
            return;
        }

        logger.info(">>>>build Jupyter NoteBook Process, command = {}", StringUtils.join(commands, " "));

        List<String> execCommand = new LinkedList<>();
        if (OSUtils.isWindows()) {
            execCommand.add(commandInterpreter());
            execCommand.addAll(commandOptions());
            execCommand.addAll(commands);

            ProcessBuilder builder = new ProcessBuilder();
            process = builder.command(execCommand).redirectErrorStream(true).start();
        } else {
            execCommand.addAll(commands);
            ProcessBuilder processBuilder = new ProcessBuilder();
            processBuilder.redirectErrorStream(true);
            processBuilder.command(execCommand);
            process = processBuilder.start();
        }

        // 获取进程ID
        Integer processId = getProcessId(process);

        // 初始化上下文/缓存任务上下文
        this.taskContext.setId(noteBookDTO.getId());
        this.taskContext.setProcessId(processId);
        this.noteBookTaskCacheManager.cacheTaskExecutionContext(this.taskContext);

        // 日志输出
        parseProcessOutput();

        logger.info("Jupyter NoteBook process start, process id is: {}", processId);

        // 等待任务执行完成(停止、手动杀死进程)
        process.waitFor();
        logger.info("Jupyter NoteBook process has exited, processId:{} ,exitStatusCode:{}", processId, process.exitValue());
        NoteBookExecuteUtils.updateNoteBookStatus(noteBookDTO, 0);
    }

    /**
     * Windows、Linux环境
     * @return
     */
    private String commandInterpreter() {
        return OSUtils.isWindows() ? CMD : SH;
    }

    /**
     * 获取执行命令可选字段
     * @return command options list
     */
    private List<String> commandOptions() {
        if (OSUtils.isWindows()) {
            return Collections.singletonList("/c");
        }else{
            return Collections.emptyList();
        }
    }

    /**
     * 获取Jupyter NoteBook进程ID
     * @param process
     * @return
     */
    private int getProcessId(Process process) {
        int processId = 0;

        try {
            Field f = process.getClass().getDeclaredField(Constants.PID);
            f.setAccessible(true);

            if (OSUtils.isWindows()) {
                WinProcess proc = new WinProcess(process);
                processId = proc.getPid();
            } else {
                processId = f.getInt(process);
            }
        } catch (Throwable e) {
            logger.error(e.getMessage(), e);
        }

        return processId;
    }

    /**
     * 处理Jupyter日志输出
     */
    private void parseProcessOutput() {
        String threadLoggerInfoName = this.taskContext.getProcessId() + "-" + "logger";
        ExecutorService parseProcessOutputExecutorService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
        parseProcessOutputExecutorService.submit(() -> {
            BufferedReader inReader = null;
            try {
                if (OSUtils.isWindows()) {
                    inReader = new BufferedReader(new InputStreamReader(process.getInputStream(), "GBK"));
                } else {
                    inReader = new BufferedReader(new InputStreamReader(process.getInputStream(), "UTF-8"));
                }

                long lastFlushTime = System.currentTimeMillis();
                String line;
                while ((line = inReader.readLine()) != null) {
                    this.taskContext.getLogBuffer().add(line);
                    lastFlushTime = flush(lastFlushTime);
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            } finally {
                clear();
                close(inReader);
            }
        });

        parseProcessOutputExecutorService.shutdown();
    }

    /**
     * 日志持久化到DB中
     * @param lastFlushTime
     * @return
     */
    private long flush(long lastFlushTime) {
        long now = System.currentTimeMillis();
        if (this.taskContext.getLogBuffer().size() >= Constants.DEFAULT_LOG_ROWS_NUM || now - lastFlushTime > Constants.DEFAULT_LOG_FLUSH_INTERVAL) {
            lastFlushTime = now;

            this.noteBookDTO.setPid(this.taskContext.getProcessId());
            NoteBookExecuteUtils.updateNoteBookLog(this.noteBookDTO, this.taskContext.getLogBuffer());
        }
        return lastFlushTime;
    }

    /**
     * 关闭流
     * @param inReader
     */
    private void close(BufferedReader inReader) {
        if (inReader != null) {
            try {
                inReader.close();
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            }
        }
    }

    /**
     * 清空日志记录
     */
    private void clear() {
        if (!this.taskContext.getLogBuffer().isEmpty()) {
            NoteBookExecuteUtils.updateNoteBookLog(this.noteBookDTO, this.taskContext.getLogBuffer());
            this.taskContext.getLogBuffer().clear();
        }
    }

    /**
     * 获取notebook服务默认用户目录
     * @return
     */
    private String getNoteBookBaseDir() {
        return ConfigUtils.getInstance().getString("phm.dds.notebook.baseDir", "/phm/jupyterdata");
    }

    /**
     * 创建notebook服务用户目录
     * @param baseUrl
     */
    private boolean makeUserDir(String baseUrl) {
        File file = new File(this.baseDir + "/" + baseUrl);
        if (file.exists()) {
            return true;
        }

        return file.mkdirs();
    }
}
